{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "349b4601-d5bd-41da-8e19-05d1d6040945",
   "metadata": {},
   "source": [
    "# Воркшоп по Kafka и Spark Structured Streaming"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4229849f-3a60-4fac-ad68-b557b937118e",
   "metadata": {},
   "source": [
    "Положить данные с Кафка в Монго с помощью Spark Structured Streaming\n",
    "\n",
    "```json\n",
    "{\n",
    "  \"client_id\": \"717651bc-cd51-11ee-b4d9-d00d11ee431a\",\n",
    "  \"timestamp\": 1708146017.469176,\n",
    "  \"lat\": 55.762747,\n",
    "  \"lon\": 37.661611\n",
    "}\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a9dd50b7-22e6-4054-be6b-d65b2406bd82",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Задание 0\n",
    "\n",
    "Все логи по умолчанию пишутся в консоль. Чтобы увидеть их в ноутбуке, необходимо выполнить следующие действия:\n",
    " - В консоли докера с `pyspark` выполнить команду `ipython profile create;`\n",
    " - В файле `.ipython/profile_default/ipython_kernel_config.py` раскомментировать строку `c.IPKernelApp.capture_fd_output = True`;\n",
    " - Перезапустить `kernel` в ноутбуке."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "26e80c44-db7d-42af-b202-8c809ac68d9d",
   "metadata": {},
   "source": [
    "## Задание 1\n",
    "\n",
    "Спроектировать пайплан. Можно нарисовать схему с базами данных, топиками Kafka и процессами Spark."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5d190d7f-2f8e-4d76-8a93-f83e55fb4511",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Задание 2\n",
    "\n",
    "Подключиться к топику с помощью `Spark DataFrame StreamReader`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ce99ea0e-9fd6-4b07-82ba-72e7f92617f4",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession, functions as F\n",
    "from pyspark.sql.types import StructType, StructField, StringType, TimestampType, DoubleType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "652ae990-ef47-4bc6-a052-cf7febf016cb",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName('yp-kafka-workshop') \\\n",
    "  .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "15d6dc6f-1b18-4826-af58-71bad9d73b61",
   "metadata": {},
   "source": [
    "Настройка `ReadStream`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b0fe6997-91b0-477b-acec-2fce4e1fa016",
   "metadata": {},
   "outputs": [],
   "source": [
    "kafka_user = 'de-student'\n",
    "kafka_pass = ''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "04eb868c-9ca6-4e02-947c-df7a8b06c92f",
   "metadata": {},
   "outputs": [],
   "source": [
    "topic_name = 'student.topic.cohort937.dtirskikh'\n",
    "\n",
    "df = spark.readStream \\\n",
    "    .format('kafka') \\\n",
    "    .option('kafka.bootstrap.servers', 'rc1b-2erh7b35n4j4v869.mdb.yandexcloud.net:9091') \\\n",
    "    .option('kafka.security.protocol', 'SASL_SSL') \\\n",
    "    .option('kafka.sasl.jaas.config', f'org.apache.kafka.common.security.scram.ScramLoginModule required username=\"{kafka_user}\" password=\"{kafka_pass}\";') \\\n",
    "    .option('kafka.sasl.mechanism', 'SCRAM-SHA-512') \\\n",
    "    .option('maxOffsetsPerTrigger', \"100\") \\\n",
    "    .option('subscribe', topic_name) \\\n",
    "    .option(\"startingOffsets\", \"earliest\") \\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bd1f64a3-8fd8-4d9b-9400-e20d298869c3",
   "metadata": {},
   "source": [
    "Проверяем загрузку данных:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "014fecbe-a178-412e-a27f-6c27576d07ab",
   "metadata": {},
   "outputs": [],
   "source": [
    "sampleQuery = df.selectExpr(\"CAST(value AS STRING)\").writeStream.format(\"console\").option(\"truncate\", False).start()\n",
    "sampleQuery.awaitTermination(20)\n",
    "sampleQuery.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2f843284-085e-4a5a-a560-4bf90eeac88f",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Задание 3\n",
    "\n",
    "Написать непосредственно преобразование данных. Это преобразование будет выполняться в функции `foreachBatch`:\n",
    "  \n",
    "Также необходимо выбрать один из триггеров: https://spark.apache.org/docs/3.1.1/api/python/reference/api/pyspark.sql.streaming.DataStreamWriter.trigger.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1c001524-d729-4d15-8a1c-787b936aea13",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Схема данных\n",
    "schema = StructType([\n",
    "    StructField(\"client_id\", StringType(), True),\n",
    "    StructField(\"timestamp\", DoubleType(), True),\n",
    "    StructField(\"lat\", DoubleType(), True),\n",
    "    StructField(\"lon\", DoubleType(), True)\n",
    "]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6de21ebf-46cc-49d3-9825-e20732e30aed",
   "metadata": {},
   "outputs": [],
   "source": [
    "mongo_config = {\n",
    "    \"connection.uri\": \"mongodb://mongodb:27017/\",\n",
    "    \"database\": \"my_database\"\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a1cf667a-416d-4b91-a8c6-a4b46c6b2f7f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Функция, которая будет выполняться в forEachBatch\n",
    "def process_data(batch_df, batch_id):\n",
    "    print(batch_df.count())\n",
    "    \"\"\" \n",
    "      Написать логику здесь:\n",
    "          1. Десереализация столбца value\n",
    "          2. Парсинг строк JSON в схему Spark\n",
    "    \"\"\"\n",
    "    res = batch_df \\\n",
    "      .select(F.col('value').cast('string')) \\\n",
    "      .select(F.from_json(F.col('value'), schema).alias('ParsedValue')) \\\n",
    "      .select(F.col('ParsedValue.*')) \\\n",
    "      .withColumn('timestamp', F.from_unixtime(F.col('timestamp'), \"yyyy-MM-dd' 'HH:mm:ss.SSS\").cast(TimestampType()))\n",
    "    \n",
    "    res.show()\n",
    "    \n",
    "    # Запись в Mongo\n",
    "    res.write \\\n",
    "      .format(\"mongodb\") \\\n",
    "      .mode(\"append\") \\\n",
    "      .option(\"collection\", \"clients\") \\\n",
    "      .options(**mongo_config) \\\n",
    "      .save()\n",
    "    \n",
    "    print(\"Batch was saved\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e225fa9d-7f7d-4637-a4dc-4bcb6eece3c0",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\" \n",
    "  Непосредственно обработка потока данных:\n",
    "    1. Определяем папку checkpoints, куда Spark будет записывать свой прогреcc\n",
    "    2. Добавляем функцию в foreachBatch\n",
    "\"\"\"\n",
    "query = df \\\n",
    "  .writeStream \\\n",
    "  .option(\"checkpointLocation\", \"file:///home/jovyan/checkpoints/query\") \\\n",
    "  .foreachBatch(process_data) \\\n",
    "  .start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "630b992d-f7c7-4815-af6c-015feba08933",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Остановить обработку:\n",
    "query.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
